package com.poetic.network.flow;

import com.poetic.network.flow.domain.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 监控温度传感器的温度值，如果温度值在一秒钟之内(processing time)
 * 连 续上升，则报警。
 */
public class ProcessFunctionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStreamSource<String> stream = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<SensorReading> dataStream = stream.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String data) throws Exception {
                String[] dataArr = data.split("\\,");
                return new SensorReading(dataArr[0].trim(), Long.valueOf(dataArr[1].trim()), Double.valueOf(dataArr[2].trim()));
            }
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(1)) {
            @Override
            public long extractTimestamp(SensorReading element) {
                // 穿过来的数据就是毫秒
                return element.getTimestamp();
            }

        });
        dataStream.keyBy(new KeySelector<SensorReading, String>() {
            @Override
            public String getKey(SensorReading value) throws Exception {
                return value.getId();
            }
        })
                .process(new KeyedProcessFunction<String, SensorReading, String>() {
            // 定义状态，保存上一次的温度
            ValueState<Double> lastTempState;

            // 定义状态，用于保存定时器的时间戳（用于删除）   
            ValueState<Long> currentTimerState;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("lastTempState",
                        TypeInformation.of(new TypeHint<Double>() {
                        })));
                currentTimerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("currentTimerState",
                        TypeInformation.of(new TypeHint<Long>() {
                        })));
            }

            @Override
            public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
                /**
                 * 1秒内温度连续上升，定义一个定时器，1秒后触发，如果温度一直高，则触发，
                 * 一旦温度低了，则移除定时器。
                 */
                // 上一次温度
                Double lastTemp = lastTempState.value() != null ? lastTempState.value() : 0.0;
                lastTempState.update(value.getTemperature());
                // 如果温度上升且没有定时器则注册定时器
                if (value.getTemperature() > lastTemp || currentTimerState.value() == null) {
                    System.out.println("------>第一次或连续升温注册定时器----");
                    // 注册定时器一秒后执行
                    // 当前时间（事件时间/处理时间）
                    long timerTs = ctx.timerService().currentProcessingTime();
                    // 不能直接传入1000（1秒），否则是1970年
//                    long time = timerTs + 1000;
                    long time = timerTs + 10000L;// 为了方便测试，这里用10秒，测试完在改为1秒
                    System.out.println("---->:" + timerTs);
                    ctx.timerService().registerProcessingTimeTimer(time);
                    // 保存定时器时间戳
                    currentTimerState.update(time);
                } else if (lastTemp > value.getTemperature() || lastTemp == 0.0) {
                    System.out.println("------>降温移除定时器----" + lastTemp);

                    // 如果温度下降，或是第一条数据
                    // 删除定时器并清空状态
                    ctx.timerService().deleteProcessingTimeTimer(currentTimerState.value());
                    currentTimerState.clear();
                } else {
                    System.out.println("----其他逻辑---上次温度：" + lastTemp + ",这次温度" + value.getTemperature() + "，定时器：" + currentTimerState.value());
                }
            }

            // 定时器回调函数
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                System.out.println("----连续升温触发任务-----");
                // 删除报警信息
                out.collect(ctx.getCurrentKey() + "温度连续上升");
                currentTimerState.clear();
            }
        }).print();
        env.execute("process demo");

    }
}
